/*
 * runner.cpp
 *
 *  Created on: 2012-7-10
 *      Author: wanghaox
 */

#include <iostream>
#include <fstream>
#include <mpi.h>
#include "base_config_file.h"
#include "base_obj.h"
#include "gbdt/gbdt.h"

using namespace std;
using namespace base;

base::Config *g_config;

static const int core_count = 3;
Updater *g_updater[core_count] = { NULL, NULL, NULL };
Syncer *g_syncer;

bool read_data(const char* file) {
	// open file, or return error
	ifstream input(file);
	if (input.fail()) {
		fprintf(stderr, "Error: unable to open training file %s\n", file);
		return false;
	}

	// process all data instances
	int linenum = 0;
	string strline;
	while (getline(input, strline)) {
		for (int i = 0; i < core_count; i++) {
			bool success = g_updater[i]->add_data(strline, linenum);
			if (!success) {
				fprintf(stderr, "Error: unable to load training file %s, line %d\n", file, linenum);
				return false;
			}
		}
		linenum++;
	}

	// indicate success
	return true;
}

int main(int argc, char** argv) {
	// step init
	FileConfig *g_file_config = new FileConfig;
	g_file_config->Init("../src/test_data/gbdt.conf");

	g_config = g_file_config;

	g_syncer = (Syncer*)get_class("GbdtSyncer");

	for (int i = 0; i < core_count; i++) {
		g_updater[i] = (Updater*)get_class("GbdtUpdater");
		g_updater[i]->_id = i;
		g_updater[i]->_count = core_count;

		g_updater[i]->init();
	}

	read_data("../src/test_data/train.dat");

	// step iteration
	int iter_num = 0;
	do {
		// updater iteration
		for (int i = 0; i < core_count; i++) {
			g_updater[i]->iteration();
		}

		int sync_num = 0;
		do {
			vector<Message*> update_messages;

			for (int i = 0; i < core_count; i++) {
				Message *update_msg = NULL;
				if (!g_updater[i]->gather_from(update_msg))
					continue;

				if (NULL == update_msg)
					continue;

				update_messages.push_back(update_msg);
			}

			Message *sync_msg = NULL;
			if (update_messages.size() > 0) {
				g_syncer->collect(update_messages, sync_msg);
			}

			if (NULL != sync_msg) {
				for (int i = 0; i < core_count; i++) {
					g_updater[i]->spread_to(sync_msg);
				}
			}

		} while (!g_syncer->is_sync_finish(sync_num++));
	} while (!g_syncer->is_end(iter_num++));

	return 0;
}
